package com.shujia.rec.funaction

import com.shujia.rec.common.Constants
import com.shujia.rec.entry.CaseClass.LogEntry
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.util.Collector
import org.apache.hadoop.hbase.CellUtil
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import redis.clients.jedis.{BitOP, Jedis}

class ItemCfSimilarityFunaction extends RichFlatMapFunction[LogEntry, LogEntry] {
  var connection: HConnection = _

  var u: HTableInterface = _
  var similarity: HTableInterface = _
  var jedis: Jedis = _

  override def open(parameters: Configuration): Unit = {
    //换件hbase连接
    val configuration = new org.apache.hadoop.conf.Configuration()
    configuration.set("hbase.zookeeper.quorum", Constants.HBASE_ZOOKEEPER_CONNECT)
    connection = HConnectionManager.createConnection(configuration)

    /**
      * create 'rec:p_history','info'
      *
      * create 'rec:u_history','info'
      *
      * create 'rec:similarity','info'
      */

    //用户购买历史表
    u = connection.getTable("rec:u_history")
    //商品相似度表
    similarity = connection.getTable("rec:similarity")
    jedis = new Jedis(Constants.REDIS_HOST, 6379)

  }

  override def flatMap(value: LogEntry, out: Collector[LogEntry]): Unit = {

    //1、获取用户购买历史
    var uSet = getHostory(String.valueOf(value.userId), u)

    //2、获取和当前物品相关的物品
    var sSet = getHostory(String.valueOf(value.proId), similarity)


    //取并集
    var unionSet = uSet | sSet

    //当前商品
    val proid = value.proId

    //自己和自己不需要计算相关性
    unionSet = unionSet.filter(id => !id.equals(String.valueOf(proid)))

    //将商品被购买的历史存到redis的位图中  共后续计算同现次数
    jedis.setbit("p_history:" + value.proId, value.userId, "1")

    //统计当前商品被购买的次数
    val j = jedis.bitcount("p_history:" + proid)

    unionSet.foreach(id => {

      //计算同现次数
      jedis.bitop(BitOP.AND, "p_history:tmp", "p_history:" + id, "p_history:" + proid)
      //同现次数
      val N = jedis.bitcount("p_history:tmp")
      jedis.del("p_history:tmp")

      //统计其他商品被购买的次数
      val i = jedis.bitcount("p_history:" + id)

      //通过余弦相似度计算物品相关性
      val score = N / Math.sqrt(i * j)

      //保存相关性到hbase
      saveSimilarity(String.valueOf(id), String.valueOf(proid), score)
      saveSimilarity(String.valueOf(proid), String.valueOf(id), score)

    })
  }

  /**
    * 保存相似度
    *
    */

  def saveSimilarity(i: String, j: String, score: Double): Unit = {
    //将相关性保存到hbase
    val put = new Put(i.getBytes())
    put.add("info".getBytes(), j.getBytes(), String.valueOf(score).getBytes())
    similarity.put(put)
  }


  /**
    * 获取用户历史和商品历史
    *
    */
  def getHostory(id: String, table: HTableInterface): Set[String] = {


    import scala.collection.JavaConverters._

    val get = new Get(id.getBytes())
    val rs = table.get(get)
    val listCell = rs.listCells()
    if (listCell == null || listCell.isEmpty) {
      return Set()
    }

    val set = listCell
      .asScala
      .toList
      .map(cell => Bytes.toString(CellUtil.cloneQualifier(cell))) //取出列名
      .toSet

    set
  }

}
